package com.gitee.theskyzero.data.kafka;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;

@RestController
@RequestMapping("/data/kafka")
public class KafkaRestController {

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    KafkaAdmin kafkaAdmin;

    @PostMapping("/message")
    public void testSendMessage(String message) {
        kafkaTemplate.send("test", message);
    }

    @GetMapping("/metrics")
    public Map<MetricName, ? extends Metric> testAdminMetrics() {
        KafkaAdminClient adminClient = (KafkaAdminClient) AdminClient.create(kafkaAdmin.getConfigurationProperties());
        return adminClient.metrics();
    }

    @KafkaListener
    public void consumeMessage(@Validated @Payload String message, @Header(KafkaHeaders.RECEIVED_TIMESTAMP) String timestamp) {
        System.out.println(message);
    }


}
